/*
 * Package il.ac.biu.cs.grossmm.test.fooAdapters
 * File FooAggregator.java
 * 
 * This is a part of presence service framework API. 
 * See javadoc for more information.
 *  
 * Copyright (C) 2006 Grossmann Mark
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public License
 * as published by the Free Software Foundation; either version 2
 * of the License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.

 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
 */
package il.ac.biu.cs.grossmm.tests.fooAdapters;

import static il.ac.biu.cs.grossmm.api.data.NodeTypeByInterface.*;
import static il.ac.biu.cs.grossmm.api.data.SetOf.*;
import static il.ac.biu.cs.grossmm.api.flow.ModificationMapper.*;
import static il.ac.biu.cs.grossmm.api.keys.PatternFactory.*;
import static il.ac.biu.cs.grossmm.api.presence.BaseVocabulary.*;
import il.ac.biu.cs.grossmm.api.OperationFailedException;
import il.ac.biu.cs.grossmm.api.data.Node;
import il.ac.biu.cs.grossmm.api.data.Nodeset;
import il.ac.biu.cs.grossmm.api.data.Property;
import il.ac.biu.cs.grossmm.api.data.Root;
import il.ac.biu.cs.grossmm.api.data.SetOf;
import il.ac.biu.cs.grossmm.api.data.UpdateType;
import il.ac.biu.cs.grossmm.api.flow.*;
import il.ac.biu.cs.grossmm.api.keys.Key;
import il.ac.biu.cs.grossmm.api.keys.KeyPattern;
import il.ac.biu.cs.grossmm.api.presence.Entity;
import il.ac.biu.cs.grossmm.api.presence.ResourceType;
import il.ac.biu.cs.grossmm.api.server.ComponentManager;
import il.ac.biu.cs.grossmm.foo.format.*;

import java.util.*;
import java.util.logging.Level;
import java.util.logging.Logger;


public class FooAggregator extends DataProcessorBase implements
Subscriber<SetOf<Presence>>, Notifier<Presence> {
	
	static Logger logger = Logger.getLogger("FooAggregator");
	
//	static {
//		logger.setLevel(Level.FINE);
//	}

	SubscriptionPoint<SetOf<Presence>> sp;

	NotificationPoint<Presence> np;

	HashMap<Key, Aggregation> aggregations = new HashMap<Key, Aggregation>();

	private static final Nodeset<SetOf<Presence>, Presence> FOOSET =
		setOf(Presence.class).NODESET;

	private static final Random random = new Random();

	static class Aggregation extends HashMap<String, Group> {

		Aggregation(NodeInspector ins, Set<Node<Presence>> foos)
			throws OperationFailedException
		{
			// for all presence documents for this presentity
			for (Node<Presence> foo : foos) {
				Set<Node<PresenceTuple>> tuples = ins.getSubnodes(foo, Presence.TUPLES);

				// for all tuples in this document
				for (Node<PresenceTuple> tuple : tuples) {

					// add the tuple to the set of tuples with the same address
					String address = ins.getValue(tuple, PresenceTuple.ADDRESS);
					Group group = get(address);

					if (group == null) {
						// set does not exist - create it
						group = new Group(address);
						put(address, group);
					}

					group.add(tuple);
				}
			}
		}

	}

	static class Group extends HashSet<Node<PresenceTuple>> {
		String address;

		Node<PresenceTuple> dstTuple;

		/**
		 * @param address
		 */
		public Group(String address)
		{
			super();
			this.address = address;
		}

		void aggregate(Worker worker) throws OperationFailedException
		{
			NodeManipulator man = worker.man;
			Node<Presence> dst = worker.dst;
			
			if( logger.isLoggable(Level.FINE) )
				logger.fine("Aggregating group for "+address);

			// is set of source tuple empty?
			if (size() == 0) {
				// yes - remove the destination tuple
				assert dstTuple != null;
				man.removeSubnode(dstTuple);
				return;
			}

			// is destination tuple already created ?
			if (dstTuple == null) {
				// no - create it
				dstTuple = man.createSubnode(dst, Presence.TUPLES);

				String id = Integer.toString(random.nextInt(), 36);
				man.setValue(dstTuple, PresenceTuple.ID, id);
			}

			// addres could change -- update it
			man.setAndCheck(dstTuple, PresenceTuple.ADDRESS, address);

			// aggregate all values for this group
			worker.aggregateValues(this);
		}
	}

	public FooAggregator()
	{
		super("Presence FooAggregator", false);
	}

	@Override
	public void setComponentManager(ComponentManager container)
		throws Exception
	{
		super.setComponentManager(container);

		KeyPattern pattern =
			pattern(Entity.class, mandatory(RESOURCE_TYPE,
				ResourceType.PRESENCE));

		// create subscription point for aggregation input
		sp = adm.createSubscriptionPoint(this, pattern, setOf(Presence.class));

		// crete notification point for aggregation output
		np = adm.createNotificationPoint(this, pattern, nodeType(Presence.class), 1);
	}

	public void activated(Key key, Root<SetOf<Presence>> srcRoot)
	{
		Root<Presence> dstRoot;

		try {
			dstRoot = np.activated(key);
		} catch (NoSuchSubscription e) {
			e.printStackTrace();
			sp.unsubscribe(key);
			aggregations.remove(key);
			return;
		}

		NodeInspector ins = sp.readLock(srcRoot);
		NodeManipulator man = np.writeLock(dstRoot);

		Worker worker = new Worker(ins, man, dstRoot);

		try {
			Aggregation a = worker.fullAggregation(srcRoot, dstRoot);
			aggregations.put(key, a);
		} catch (OperationFailedException e) {
			e.printStackTrace();
			sp.unsubscribe(key);
			aggregations.remove(key);
		} finally {
			np.writeUnlock(dstRoot);
			sp.readUnlock(srcRoot);
		}
	}

	static class Worker {

		NodeInspector ins;
		NodeEventInspector eins;

		NodeManipulator man;

		Root<Presence> dst;

		/**
		 * @param ins
		 * @param man
		 * @param dst
		 */
		public Worker(NodeInspector ins, NodeManipulator man, Root<Presence> dst)
		{
			super();
			this.ins = ins;
			this.man = man;
			this.dst = dst;
			
			if( ins instanceof NodeEventInspector )
				eins = (NodeEventInspector) ins;
		}

		void aggMood(Set<Node<Presence>> set) throws OperationFailedException
		{
			int h = 0, c = 0, w = 0, l = 0;
			int count = 0;
			for (Node<Presence> node : set) {
				Mood mood = ins.getValue(node, Presence.MOOD);
				if (mood == null)
					continue;
				h += mood.happiness;
				c += mood.calmness;
				w += mood.weariness;
				l += mood.loneliness;

				count++;
			}

			if (count == 0) {
				man.setAndCheck(dst, Presence.MOOD, null);
			} else {
				Mood newMood = new Mood(h/count, c/count, w/count, l/count);
				man.setAndCheck(dst, Presence.MOOD, newMood);
			}
		}

		void aggActivities(Set<Node<Presence>> set) throws OperationFailedException
		{
			EnumSet<Activity> acs = EnumSet.noneOf(Activity.class);

			for (Node<Presence> node : set) {
				Activities a = ins.getValue(node, Presence.ACTIVITIES);
				acs.addAll(a);
			}

			man.setAndCheck(dst, Presence.ACTIVITIES, new Activities(acs));
		}
		
		void aggEntity(Set<Node<Presence>> set) throws OperationFailedException
		{
			findNonNull(dst, set, Presence.ENTITY);
		}


		void aggSphere(Set<Node<Presence>> set) throws OperationFailedException
		{
			aggAllEqual(dst, set, Presence.SPHERE);
		}

		void aggLocation(Set<Node<Presence>> set) throws OperationFailedException
		{
			aggAllEqual(dst, set, Presence.LOCATION);
		}

		/**
		 * @param dst2
		 * @param set
		 * @param property
		 * @throws OperationFailedException
		 */
		private <N, S> void aggAllEqual(Node<N> node, Set<Node<N>> set,
			Property<N, S> property) throws OperationFailedException
		{
			if (set.size() == 0) {
				man.setAndCheck(node, property, null);
				return;
			}

			Iterator<Node<N>> it = set.iterator();
			S val = ins.getValue(it.next(), property);

			while (it.hasNext()) {
				if (!same(val, ins.getValue(it.next(), property))) {
					man.setAndCheck(node, property, null);
					return;
				}
			}

			man.setAndCheck(node, property, val);
		}
		
		/**
		 * @param dst2
		 * @param set
		 * @param property
		 * @throws OperationFailedException
		 */
		private <N, S> void findNonNull(Node<N> node, Set<Node<N>> set,
			Property<N, S> property) throws OperationFailedException
		{
			if (set.size() == 0) {
				man.setAndCheck(node, property, null);
				return;
			}

			Iterator<Node<N>> it = set.iterator();
			S val = null;
			while (it.hasNext()) {
				val = ins.getValue(it.next(), property);

				if( val != null )
					break;
			}

			man.setAndCheck(node, property, val);

		}

		void aggCaps(Group group)
			throws OperationFailedException
		{
			Set<String> caps = new HashSet<String>();

			for (Node<PresenceTuple> node : group ) {
				Capabilities c = ins.getValue(node, PresenceTuple.CAPABILITIES);
				if( c!=null )
					caps.addAll(c);
			}

			man.setAndCheck(group.dstTuple, PresenceTuple.CAPABILITIES, Capabilities.of(caps));
		}

		void aggClass(Group group)
			throws OperationFailedException
		{
			aggAllEqual(group.dstTuple, group, PresenceTuple.CLASS);
		}

		void aggPriority(Group group)
			throws OperationFailedException
		{
			float f = 0f;

			for (Node<PresenceTuple> srcTuple : group)
				f += ins.getValue(srcTuple, PresenceTuple.PRIORITY);

			man.setAndCheck(group.dstTuple, PresenceTuple.PRIORITY, f / group.size());
		}

		void aggStatus(Group group)
			throws OperationFailedException
		{
			OnlineStatus status = OnlineStatus.OFFLINE;

			for (Node<PresenceTuple> srcTuple : group) {
				OnlineStatus s = ins.getValue(srcTuple, PresenceTuple.STATUS);

				if (s.ordinal() < status.ordinal())
					status = s;
			}

			man.setAndCheck(group.dstTuple, PresenceTuple.STATUS, status);
		}

		private Aggregation fullAggregation(Root<SetOf<Presence>> srcRoot,
			Root<Presence> dstRoot) throws OperationFailedException
		{
			Set<Node<Presence>> foos = ins.getSubnodes(srcRoot, FOOSET);

			// create aggregation groups
			Aggregation aggregation = new Aggregation(ins, foos);

			// aggregate all groups
			applyAggregation(aggregation);

			// aggregate document root values
			aggregateValues(foos);

			// return created aggregation
			return aggregation;
		}

		private void fullReaggregation(Aggregation aggregation,
			Set<Node<Presence>> foos) throws OperationFailedException
		{
			// create aggregation groups
			Aggregation newAggregation = new Aggregation(ins, foos);

			// remove dst tuples which are in the old aggregation but not in the
			// new
			for (String address : aggregation.keySet()) {
				Group newGroup = newAggregation.get(address);
				if ( newGroup == null ) {

					// get group for a removed address
					Group group = aggregation.get(address);

					// remove destination touple for this address
					man.removeSubnode(group.dstTuple);
					
					if( logger.isLoggable(Level.FINE) )
						logger.fine("Removed dst tuple for "+address);
				} else {
					Group group = aggregation.get(address);
					newGroup.dstTuple = group.dstTuple;
				}
			}

			// aggregate all groups
			applyAggregation(newAggregation);
			
			aggregateValues(foos);
		}

		/**
		 * @param aggregation
		 * @throws OperationFailedException
		 */
		private void applyAggregation(Aggregation aggregation)
			throws OperationFailedException
		{
			// aggregate tuples
			for (Group group : aggregation.values()) {
				group.aggregate(this);
			}
		}

		public void smartAggregation(Group group) throws OperationFailedException
		{
			// aggregate only on properties which has changed in
			// at least one modified tuple
			Property[] properties = getModifiedProps(group, eins);
			
			for( Property p : properties ) {
				if( p == PresenceTuple.PRIORITY )
					aggPriority(group);
				if( p == PresenceTuple.STATUS )
					aggStatus(group);
				if( p == PresenceTuple.CLASS )
					aggClass(group);
				if( p == PresenceTuple.CAPABILITIES )
					aggCaps(group);
			}
		}
		
		public void smartAggregation(Set<Node<Presence>> foos) throws OperationFailedException
		{
			// aggregate only on properties which has changed in
			// at least one modified tuple
			Property[] properties = getModifiedProps(foos, eins);
			
			for( Property p : properties ) {
				if( p == Presence.ENTITY)
					aggEntity(foos);
				if( p == Presence.ACTIVITIES )
					aggActivities(foos);
				if( p == Presence.LOCATION )
					aggLocation(foos);
				if( p == Presence.MOOD )
					aggMood(foos);
				if( p == Presence.SPHERE )
					aggSphere(foos);
			}
		}

		/**
		 * @param foos
		 * @throws OperationFailedException
		 */
		private void aggregateValues(Set<Node<Presence>> foos)
			throws OperationFailedException
		{
			aggEntity(foos);
			aggActivities(foos);
			aggLocation(foos);
			aggMood(foos);
			aggSphere(foos);
		}

		/**
		 * @param dstTuple
		 * @param group
		 * @throws OperationFailedException
		 */
		private void aggregateValues(Group group)
			throws OperationFailedException
		{
			aggCaps(group);
			aggClass(group);
			aggPriority(group);
			aggStatus(group);
		}

		static boolean same(Object a, Object b)
		{
			if (a == null)
				return b == null;

			return a == b || a.equals(b);
		}

	}

	public void deactivated(Key key, Status status) throws NoSuchSubscription
	{
		aggregations.remove(key);
		np.deactivated(key, status);
	}

	public void notify(Root<SetOf<Presence>> srcRoot, NodeEventInspector ins)
	{
		if( logger.isLoggable(Level.FINE) ) {
			logger.fine(ins.toString(srcRoot));
		}
		
		Key key = srcRoot.getKey();
		Aggregation aggregation = aggregations.get(key);

		Root<Presence> dstRoot = np.getRoot(key);

		NodeManipulator man = np.writeLock(dstRoot);
		try {
			aggregate(srcRoot, dstRoot, ins, man, aggregation);
		} catch (OperationFailedException e) {
			e.printStackTrace();
			sp.unsubscribe(key);
			try {
				np.deactivated(key, Status.ERROR);
			} catch (NoSuchSubscription e1) {
				// TODO Auto-generated catch block
				e1.printStackTrace();
			}
		} finally {
			np.writeUnlock(dstRoot);
		}
	}

	/**
	 * @param srcRoot
	 * @param dstRoot
	 * @param ins
	 * @param man
	 * @param aggregation
	 * @throws OperationFailedException
	 */
	private void aggregate(Root<SetOf<Presence>> srcRoot, Root<Presence> dstRoot,
		NodeEventInspector ins, NodeManipulator man, Aggregation aggregation)
		throws OperationFailedException
	{
		Worker worker = new Worker(ins, man, dstRoot);

		// are there added or removed published documents for this presentity?
		if (ins.getSubnodes(srcRoot, FOOSET, UpdateType.ADDED).size() != 0
			|| ins.getSubnodes(srcRoot, FOOSET, UpdateType.REMOVED).size() != 0) {
			
			if( logger.isLoggable(Level.FINE) )
				logger.fine("Performing full reaggregation");

			Set<Node<Presence>> foos = ins.getSubnodes(srcRoot, FOOSET);

			// if yes - reaggregate everything
			worker.fullReaggregation(aggregation, foos);

			return;
		}

		// get set of changed documents
		Set<Node<Presence>> chFoos =
			ins.getSubnodes(srcRoot, FOOSET, UpdateType.CHANGED);

		List<Group> groups = null;

		// are there any added or removed tuples?
		for (Node<Presence> foo : chFoos) {
			Set<Node<PresenceTuple>> added =
				ins.getSubnodes(foo, Presence.TUPLES, UpdateType.ADDED);
			Set<Node<PresenceTuple>> removed =
				ins.getSubnodes(foo, Presence.TUPLES, UpdateType.REMOVED);

			for (Node<PresenceTuple> ft : added) {
				if (groups == null)
					groups = new ArrayList<Group>();

				String address = ins.getValue(ft, PresenceTuple.ADDRESS);
				Group group = aggregation.get(address);
				if (group == null) {
					group = new Group(address);
					aggregation.put(address, group);
				}

				group.add(ft);
				groups.add(group);
				
				if( logger.isLoggable(Level.FINE) )
					logger.fine("Added group for "+address);
			}

			for (Node<PresenceTuple> ft : removed) {
				if (groups == null)
					groups = new ArrayList<Group>();

				String address = ins.getValue(ft, PresenceTuple.ADDRESS);
				Group group = aggregation.get(address);
				assert group != null;

				group.remove(ft);
				groups.add(group);
				
				if( logger.isLoggable(Level.FINE) )
					logger.fine("Removed group for "+address);
			}
			
			worker.smartAggregation(chFoos);
		}

		if (groups != null) {
			if( logger.isLoggable(Level.FINE) )
				logger.fine("Got moved tuples");
			
			for (Group group : groups)
				group.aggregate(worker);
		}

		// go through all changed (previously existied and not removed) tuples
		// of all documents
		for (Node<Presence> foo : chFoos) {
			Set<Node<PresenceTuple>> changed =
				ins.getSubnodes(foo, Presence.TUPLES, UpdateType.CHANGED);

			for (Node<PresenceTuple> ft : changed) {
				String address = ins.getValue(ft, PresenceTuple.ADDRESS);
				Group group = aggregation.get(address);
				assert group != null;
				
				if( group == null ) {
					logger.warning("Workaround...............");
					Set<Node<Presence>> foos = ins.getSubnodes(srcRoot, FOOSET);
					worker.fullReaggregation(aggregation, foos);
					return;
				}

				// smart-aggregate groups, skipping already aggregated groups 
				if (groups == null || !groups.contains(group)) {
					if( logger.isLoggable(Level.FINE) )
						logger.fine("Smart group aggregation for "+address);
					worker.smartAggregation(group);
					
					if (groups == null)
						groups = new ArrayList<Group>();
					groups.add(group);
				}
			}
		}
	}

	public void fetch(Key key, Track track, Root<Presence> node,
		NodeManipulator manipulator)
	{
		// TODO Implement fetch
		// 
		throw new RuntimeException("Not implemented");
	}

	public boolean subscribe(Key key, Track track) throws Exception
	{
		Status s = sp.subscribe(key, track);
		return s != Status.UNMATCHED;
	}

	public void unsubscribe(Key key)
	{
		// TODO Implement unsubscribe
		// 
		throw new RuntimeException("Not implemented");
	}

}
